package service

import (
	"fmt"
	"github.com/Shopify/sarama"
	"log"
)

func main() {
	brokers := []string{ "47.100.122.212:30040" }
	topic := "og-visual-test"

	res := make(chan string)
	//groupId := "og-visual-test"

	go KafkaConsumer(brokers, topic, res)

	for {
		select {
		case msg := <- res :

			fmt.Println(msg)
		}
	}
}

func KafkaConsumer(brokers []string, topic string, res chan string)  {
	config := sarama.NewConfig()

	client, err := sarama.NewClient(brokers, config)
	if err != nil {
		log.Fatalf("unable to create kafka client: %q", err)
	}

	consumer, err := sarama.NewConsumerFromClient(client)
	if err != nil {
		log.Fatal(err)
	}
	defer consumer.Close()

	consume(consumer, topic, res)
}


func consume(consumer sarama.Consumer, topic string, res chan string) {
	partitions, _ := consumer.Partitions(topic)
	for i := 0; i < len(partitions); i++ {
		partition := partitions[i]
		partitionConsumer, err := consumer.ConsumePartition(topic, partition, sarama.OffsetNewest)
		if err != nil {
			log.Println(err)
			return
		}
		go loopConsume(partitionConsumer, partition, res)
	}
}

func loopConsume(partitionConsumer sarama.PartitionConsumer, partition int32, res chan string) {
	defer partitionConsumer.Close()

	for {
		msg := <-partitionConsumer.Messages()
		if msg == nil {
			fmt.Println("msg is nil")
			continue
		}

		res <- string(msg.Value)
		//log.Printf("Consumed message: [%s], partition: [%d], offset: [%d]\n", msg.Value, partition, msg.Offset)
	}
}